package com.atguigu.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * create by undeRdoG on  2021-06-19  18:02
  * 凡心所向，素履以往，生如逆旅，一苇以航。
  */
object SparkStreaming06_Join {

  def main(args: Array[String]): Unit = {
    // 创建环境
    /**
      * StreamingContext  创建时需要两个参数
      * 第一个参数环境配置
      * 第二个参数 批处理周期  duration
      **/
    val sparkConf = new SparkConf().setAppName("Streaming").setMaster("local[*]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))


    // 获取端口数据
    val data9999: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 9999)
    val data8888: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 8888)


    val map9999: DStream[(String, Int)] = data9999.map((_,9))
    val map8888: DStream[(String, Int)] = data8888.map((_,8))


    /**
    *    Dstream 的 join  本质上就是 rdd的join
    * */
    val value: DStream[(String, (Int, Int))] = map9999.join(map8888)

    value.print()

    // 启动采集器
    ssc.start()

    // 等待采集器关闭
    ssc.awaitTermination()

  }

}
